from kafka import KafkaConsumer, TopicPartition

consumer = KafkaConsumer("test_monitor_tendency",
                         group_id="test_monitor_tendency_tendency2kfk",
                         bootstrap_servers="hadoop000:9092")
partitions = [TopicPartition("test_monitor_tendency", p) for p in consumer.partitions_for_topic("test_monitor_tendency")]

print("start to cal offset:")

# total
toff = consumer.end_offsets(partitions)
toff = [(key.partition, toff[key]) for key in toff.keys()]
toff.sort()
# print("total offset: {}".format(str(toff)))

# current
coff = [(x.partition, consumer.committed(x)) for x in partitions]
coff.sort()
# print("current offset: {}".format(str(coff)))

# cal sum and left
toff_sum = sum([x[1] for x in toff])
cur_sum = sum([x[1] for x in coff if x[1] is not None])
left_sum = toff_sum - cur_sum
print("topic:" + "test_monitor_tendency" + ",kafka left: {}".format(left_sum))